/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity;

import java.math.BigDecimal;
import java.math.RoundingMode;

import org.junit.Test;
import org.sabayframework.activity.impl.Fork;
import org.sabayframework.activity.impl.Pipeline;
import org.sabayframework.activity.impl.ReplicableActivity;
import org.sabayframework.activity.impl.StatefulActivity;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.ram.ArrayChannel;
import org.sabayframework.channel.ram.ConcurrentArrayChannel;
import org.sabayframework.queue.ExecutionQueue;
import org.sabayframework.queue.impl.CentralizedExecutionQueue;

/**
 * Calculation of Pi using Leibniz formula.
 * http://en.wikipedia.org/wiki/Leibniz_formula_for_%CF%80
 * @author Alex Antonau
 * */
public class LeibnizPi {

	static class Series{
		private int start;
		private int length;
	}
		
	public static BigDecimal calculate(final int startN,final int nrOfElements) {
	      BigDecimal acc = BigDecimal.ZERO;
	      final int limitN = startN + nrOfElements;
	      for (int i = startN; i < limitN; i++) {
	    	final BigDecimal a = new BigDecimal(1 - (i % 2) * 2);
	    	final BigDecimal b = new BigDecimal(2 * i + 1);
	    	acc = acc.add(a.divide(b, 30, RoundingMode.HALF_UP));
	      }
	      final BigDecimal four = new BigDecimal(4);
	      return acc.multiply(four);
	}
	
	static class Calculator extends ReplicableActivity<Series, BigDecimal>{

		private BigDecimal result;
		private Series series;
		@Override
		public ReplicableActivity<Series, BigDecimal> replicate() {
			return new Calculator();
		}

		public void input(ActivityContext<Series, BigDecimal> ctx, Lock<Series> inputLock) throws Exception {
			series = inputLock.get(0);
		}

		@Override
		public void process(ActivityContext<Series, BigDecimal> ctx){
			result = calculate(series.start, series.length);
		}

		public void output(ActivityContext<Series, BigDecimal> ctx,	Lock<BigDecimal> outputLock) throws Exception {
			outputLock.set(0, this.result);
		}
	}
	
	static class Summator extends StatefulActivity<BigDecimal, BigDecimal>{
		
		private BigDecimal accumulator=BigDecimal.ZERO;
		
		public void afterExecute(ActivityContext<BigDecimal, BigDecimal> ctx) throws Exception {
			ctx.getOutput().add(accumulator);
			super.afterExecute(ctx);
		}

		public void input(ActivityContext<BigDecimal, BigDecimal> ctx, Lock<BigDecimal> inputLock) throws Exception {
			BigDecimal n = inputLock.get(0); 
			accumulator = accumulator.add(n);
		}
		public int getOutputSize(ActivityContext<BigDecimal, BigDecimal> ctx) {
			return 0;
		}
		public void output(ActivityContext<BigDecimal, BigDecimal> ctx,	Lock<BigDecimal> outputLock)  {}
	}
	
	public static BigDecimal calculate(int seriesLength) throws Exception{
		final CentralizedExecutionQueue fs = new CentralizedExecutionQueue();
		return calculate(fs, seriesLength);
	}
	public static BigDecimal calculate(ExecutionQueue fs, int seriesLength) throws Exception{
		Pipeline<Series,BigDecimal> pipe= new Pipeline<Series,BigDecimal>(
				new Fork<Series, BigDecimal>(new Calculator()), 
				new Summator());
		
		int partitions = fs.workerCount();
		Channel<Series> input=new ArrayChannel<Series>(partitions);
		int blockLength = seriesLength / partitions;
		int reminder = seriesLength % partitions;
		for(int i=0;i<partitions;++i){
			Series s = new Series(); 
			s.start = blockLength * i;
			s.length = blockLength;
			if(i==partitions-1) s.length+=reminder;
			input.add(s);
		}
		
		input.close();
		Channel<BigDecimal> output = new ConcurrentArrayChannel<BigDecimal>(1);
		return fs.execute(pipe, input, output).waitForResult(5000000);
	}
	private static int N = 5000000;
	
	@Test
	public void testSerial(){
		calculate(0, N);
		long startTime = System.currentTimeMillis();
		System.out.println(calculate(0, N));
		long endTime = System.currentTimeMillis();
		long actualTime = endTime - startTime;
		System.out.println("Serial Leibniz time,ms = "+actualTime);
		
	}
	
	@Test
	public void testParallel() throws Exception {
		calculate(N);
		long startTime = System.currentTimeMillis();
		System.out.println(calculate(N));
		long endTime = System.currentTimeMillis();
		long actualTime = endTime - startTime;
		System.out.println("Parallel Leibniz time,ms = "+actualTime);
	}

}
